package com.zte.claa.inficombo.csa.app.service.rptparse.ext.fengyu;

import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import com.zte.claa.inficombo.csa.app.config.FengYuSvrCommProps;
import com.zte.claa.inficombo.csa.app.dao.mysql.impl.DevBaseMapper;
import com.zte.claa.inficombo.csa.app.model.dev.DevBase;
import com.zte.claa.inficombo.csa.app.model.dev.DevTypeModel;
import com.zte.claa.inficombo.csa.app.model.redis.AlarmData;
import com.zte.claa.inficombo.csa.app.model.redis.DevMeteData;
import com.zte.claa.inficombo.csa.app.service.csa.impl.BaseCacheManager;
import com.zte.claa.inficombo.csa.app.service.rptparse.ParserContainer;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.ExtRptDataParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.RptRawData;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.CommParamProps;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.CommService;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.CommServiceImpl;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.mina.msgbean.MinaCommRequest;
import com.zte.claa.inficombo.csa.app.util.JsonUtils;
import com.zte.claa.inficombo.csa.app.util.SecurityUtil;

/**
 * 第三方需要透传原始数据的的枫煜电气火灾服务实现类。
 * 
 * @author 0283000115
 */
@Service("FengYuParserService")
@Lazy(false)
public class FengYuParserService implements InitializingBean, DisposableBean, ExtRptDataParser {	
	// 缓存队列大小
	private static final int QUEUE_MAX_SIZE = 500;
	// 日志记录
	private final Logger logger = LoggerFactory.getLogger(FengYuParserService.class);
	
	@Autowired
	private ParserContainer parserContainer;
	
	@Autowired
	private FengYuSvrCommProps commProps;
	
	@Autowired
	private BaseCacheManager cacheManager;
	
	@Autowired
	private DevBaseMapper devMapper;
	
	// 内部分发队列...
	private Thread dataDealear = null;
	private LinkedBlockingQueue<RptRawData> queue = null;
	private boolean isWorkOn = true;
	
	// 
	private CommService destSvr = null;
	
	/**
	 * 服务初始化.
	 */
	@Override
	public void afterPropertiesSet() throws Exception {
		// 注册原始数据发送处理器.
		String devtp = this.commProps.getDevtype().trim();
		this.parserContainer.setExtRptDataDealer(devtp, this);
		logger.info("setExtRptDataDealer for devtp ok: " + devtp);
		// 创建commsvr对象，并不实际建立连接		
		this.destSvr = new CommServiceImpl(this.transferCommProps());
		// 共用一个缓冲队列.
		this.queue = new LinkedBlockingQueue<RptRawData>(QUEUE_MAX_SIZE);
		this.isWorkOn = true;
		this.dataDealear = new Thread(new Runnable() {
			@Override
			public void run() {
				dealRptRawData();
			}
		});
		this.dataDealear.setName("FengYuParserService-dispatcher");
		this.dataDealear.start();
		logger.info("FengYuParserService is inited ok!");
	}
	
	/**
	 * 退出销毁资源.
	 */
	@Override
	public void destroy() throws Exception {
		this.isWorkOn = false;
		if(this.queue != null){
			this.queue.clear();
			this.queue.offer(new RptRawData(RptRawData.INVALID_DEVEUI));
		}		
		// 关闭连接
		if(this.destSvr != null){
			this.destSvr.closeConnectToDestSvr();
		}
		logger.warn("Service exit!");
	}
	
	/**
	 * 
	 * @return
	 */
	private CommParamProps transferCommProps() {
		CommParamProps cpp = new CommParamProps();
		cpp.setBufsize(this.commProps.getBufsize());
		cpp.setCharsetname(this.commProps.getCharsetname());
		cpp.setCmdsendtimeoutsecond(this.commProps.getCmdsendtimeoutsecond());
		cpp.setCmdtimeoutsecond(this.commProps.getCmdtimeoutsecond());
		cpp.setConnwaitsecond(this.commProps.getConnwaitsecond());
		cpp.setPort(this.commProps.getPort());
		cpp.setSessionidlesecond(this.commProps.getSessionidlesecond());
		cpp.setSvrip(this.commProps.getSvrip());
		return cpp;
	}

	/**
	 * 分发采集器上报的原始码流...
	 */
	@Override
	public void processRptData(RptRawData data) throws Exception {
		// 添加到linkedqueue中，等待解析处理...
		boolean b = this.queue.offer(data);
		if(!b){
			logger.warn("[*!!*]Add data to queue failed: "+data.toString());
		}
	}
	
	/**
	 * 分配处理数据.
	 */
	private void dealRptRawData() {
		while (this.isWorkOn) {
			try {
				// 发送原始码流到Dest服务器进行数据转换...
				RptRawData rawData = this.queue.poll(120, TimeUnit.SECONDS);
				if(rawData == null){
					logger.debug("No raw data available now!");
					continue;
				} 
				if(RptRawData.INVALID_DEVEUI.equalsIgnoreCase(rawData.getDeveui())){
					continue;
				}
				byte[] bs = rawData.getPayload();
				bs = (bs == null) ? new byte[0] : bs;
				logger.info("Begin submit raw data to destSvr: "+this.commProps.getSvrip()+this.commProps.getPort()+", size="+bs.length+", data="+SecurityUtil.parseByte2HexStr(bs));
				MinaCommRequest req = new MinaCommRequest();
				req.setRawData(rawData);
				this.destSvr.execCommandIngoreRsp(req);
				try{
					Thread.sleep(2);
				}catch(Throwable ignore){}
			} catch (Throwable t) {
				// 这里做一个保护，如果发送失败，则尝试关闭到目标服务器的链路。注意不能赋值为null.
				try {
					this.destSvr.closeConnectToDestSvr();
				} catch (Exception e) {
					logger.error(e.getMessage(), e);
				}
				logger.error("dealRptRawData failed: " + t.getMessage(), t);
			}
		}
		logger.warn("Dealer thread exit: " + Thread.currentThread().getName());
	}

	/**
	 * 异步接收destsvr的解析结果，持久化数据并发布.
	 */
	public void onParseMeteDataReceive(FengYuParseResult data) {
		if(data == null){
			logger.error("[*!!*]result data is null!");
			return;			
		}
		// 根据和枫煜电气火灾设备厂家约定，这里sensorno=deveui.
		DevBase dev = this.cacheManager.getDev(data.getSensorno());
		if(dev == null){
			logger.error("[*@@*]Can not get dev data for sensor: "+data.getSensorno());
			return;	
		}
		DevTypeModel dtm = this.cacheManager.getDevTypeModel(dev.getDevType());
		if(dtm == null || dtm.getDmList() == null){
			logger.error("[*!!*]Can not get datamodel for devtype="+dev.getDevType());
			return;
		}
		Map<String, Object> mdata = JsonUtils.beanToHashMap(data);
		DevMeteData meteData = new DevMeteData();
		meteData.setAppeui(dev.getAppEui());
		meteData.setDeveui(dev.getDevEui());
		meteData.setDevname(dev.getDevName());
		meteData.setDevtype(dev.getDevType());
		meteData.setPort("0");
		meteData.setProjectid(dev.getProjectId());
		meteData.setRpttime(data.parseCreatetime());
		// 删除模型无关字段...
		mdata.remove(FengYuParseResult.CREATETIME_F_NAME);
		mdata.remove(FengYuParseResult.SENSOR_F_NAME);
		meteData.setMdata(mdata);
		// 持久化数据.
		try {
			this.parserContainer.getPersisService().saveDevMeteData(meteData, dtm.getBase().getTbname());
			// 构造广播数据，进行redis发布...
			String pubMsg = JsonUtils.toJson(meteData);
			this.parserContainer.getUpdataPublisher().publishMessage(pubMsg);
			logger.info("[*^_^*]Publish parse result data ok: " + pubMsg);
		}catch(Exception ex){
			logger.error("onParseResultReceive process error: "+data, ex);
		}
	}
	
	/**
	 * 异步接收destsvr的解析结果，持久化数据并发布.
	 */
	public void onParseAlarmReceive(AlarmData alarm) {
		logger.debug("Receive alarm: " + alarm.toString());
		String alarmJson = JsonUtils.toJson(alarm);
		this.parserContainer.getAlarmPublisher().publishMessage(alarmJson);
		logger.info("[*^_^*]Publish alarm message ok: " + alarmJson);
	}
	
	/**
	 * 异步接收destsvr的解析结果，持久化数据并发布.
	 */
	public void onParseDevAddrReceive(String devEui, String devAddr) {
		logger.debug("Receive dev address: " + devEui + ", addr="+devAddr);
		try{
			int retcode = this.devMapper.updateDevAddress(devEui, devAddr);
			logger.info("Update dev address: "+devEui+", addr="+devAddr+", retcode="+retcode);
		}catch(Exception ex){
			logger.error("onParseDevAddrReceive failed: "+devEui+", addr="+devAddr, ex);
		}
	}
	
	/**
	 * 后门功能，手工关闭到服务器的连接...
	 */
	public void closeConnectionToDestSvr(){
		if(this.destSvr != null){
			this.destSvr.closeConnectToDestSvr();
		}
	}
}
